package priv.tuyou.myself;

import java.io.PrintWriter;
import java.util.Properties;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;

public class EventTimeConsumer {

	public static void main(String[] args) throws Exception {
		
		PrintWriter received = new PrintWriter("receivd.txt");
		
		Properties properties = new Properties();
//		properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getCanonicalName());// key反序列化方式
//		properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getCanonicalName());// value反系列化方式
//		properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);// 提交方式
		properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.100.124.205:9092");// 指定broker地址，来找到group的coordinator
		properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");// 指定用户组
		FlinkKafkaConsumer09<String> kafkaSource = new FlinkKafkaConsumer09<>("test2", new SimpleStringSchema(), properties);
//		kafkaSource.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<String>() {
//
//		    @Override
//		    public long extractAscendingTimestamp(String element) {
//		        return Long.parseLong(element);
//		    }
//		});
		kafkaSource.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<String>() {
			
			private final long maxOutOfOrderness = 3500; // 3.5 seconds
		    private long currentMaxTimestamp;
			
			@Override
			public long extractTimestamp(String element, long previousElementTimestamp) {
				long timestamp = Long.parseLong(element);
		        currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
		        received.println(element);
		        received.flush();
		        return timestamp;
			}
			
			@Override
			public Watermark getCurrentWatermark() {
				return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
			}
		});
//		kafkaSource.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<String>() {
//			
//			private final long maxTimeLag = 5000; // 5 seconds
//
//			@Override
//			public long extractTimestamp(String element, long previousElementTimestamp) {
//				return Long.parseLong(element);
//			}
//			
//			@Override
//			public Watermark getCurrentWatermark() {
//				return new Watermark(System.currentTimeMillis() - maxTimeLag);
//			}
//		});
//		kafkaSource.assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks<String>() {
//			
//			@Override
//			public long extractTimestamp(String element, long previousElementTimestamp) {
//				return Long.parseLong(element);
//			}
//			
//			@Override
//			public Watermark checkAndGetNextWatermark(String lastElement, long extractedTimestamp) {
//				return true ? new Watermark(extractedTimestamp) : null;
//			}
//		});

		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
		env.enableCheckpointing(5000);
		env.addSource(kafkaSource)
				.flatMap(new FlatMapFunction<String, Tuple2<Long, String>>() {
					@Override
					public void flatMap(String value, Collector<Tuple2<Long, String>> out) throws Exception {
						out.collect(new Tuple2<>(Long.parseLong(value), value));
					}
				})
				.keyBy(0)
				.timeWindow(Time.seconds(5))
				.max(0)
				.writeAsText("segment.txt");
		env.execute();
	}
	
	public static class MyType{
		private long timestamp;
		private String value;
		
		public MyType(long timestamp, String value) {
			super();
			this.timestamp = timestamp;
			this.value = value;
		}

		public long eventTimestamp(){
			return timestamp;
		}

		public String getValue() {
			return value;
		}
	}
}
